import sqlite3
import os
import pandas as pd
from datetime import datetime
import threading
import concurrent.futures
import logging
import time

# 配置日志
logger = logging.getLogger(__name__)


class StockCacheManager:
    """线程安全的股票缓存管理类"""
    _instance = None
    _lock = threading.Lock()
    _thread_local = threading.local()
    # 创建线程池执行器（最大18个线程）
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=18)
    # 添加类级锁，确保数据库操作线程安全
    _db_lock = threading.Lock()

    def __new__(cls):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super(StockCacheManager, cls).__new__(cls)
                # 创建缓存目录
                cls._instance.CACHE_DIR = "stock_cache"
                if not os.path.exists(cls._instance.CACHE_DIR):
                    os.makedirs(cls._instance.CACHE_DIR)

                # 数据库文件路径
                cls._instance.DB_FILE = os.path.join(
                    cls._instance.CACHE_DIR, "stock_cache.db"
                )
        return cls._instance

    def _get_connection(self):
        """为每个线程获取独立的数据库连接"""
        if not hasattr(self._thread_local, "conn") or self._thread_local.conn is None:
            # 允许不同线程使用同一个连接
            self._thread_local.conn = sqlite3.connect(
                self.DB_FILE,
                check_same_thread=False
            )
            self._thread_local.conn.execute("PRAGMA journal_mode=WAL")  # 启用WAL模式提高并发性能
            self.create_tables(self._thread_local.conn)
        return self._thread_local.conn

    def create_tables(self, conn):
        """创建数据库表"""
        cursor = conn.cursor()

        # 创建股票元数据表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS stock_metadata (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_input TEXT NOT NULL,
            mapped_code TEXT NOT NULL,
            stock_name TEXT NOT NULL,
            last_updated TEXT NOT NULL,
            access_count INTEGER DEFAULT 0,
            last_accessed TEXT
        )
        ''')

        # 创建K线数据表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS stock_data (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            stock_id INTEGER NOT NULL,
            date TEXT NOT NULL,
            open REAL NOT NULL,
            high REAL NOT NULL,
            low REAL NOT NULL,
            close REAL NOT NULL,
            volume INTEGER,
            FOREIGN KEY (stock_id) REFERENCES stock_metadata(id)
        )
        ''')

        # 创建关键点表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS key_points (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            stock_id INTEGER NOT NULL,
            date TEXT NOT NULL,
            point_type TEXT NOT NULL,
            price REAL NOT NULL,
            FOREIGN KEY (stock_id) REFERENCES stock_metadata(id)
        )
        ''')

        # 创建自选股表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS watchlist (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            stock_id INTEGER NOT NULL UNIQUE,
            added_time TEXT DEFAULT (datetime('now')),
            FOREIGN KEY (stock_id) REFERENCES stock_metadata(id)
        )
        ''')

        conn.commit()

    def get_stock_id(self, user_input, mapped_code, stock_name):
        """获取或创建股票ID"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()

            # 检查是否已存在
            cursor.execute('''
            SELECT id, access_count FROM stock_metadata 
            WHERE user_input = ? AND mapped_code = ?
            ''', (user_input, mapped_code))

            result = cursor.fetchone()

            if result:
                stock_id = result[0]
                access_count = result[1] + 1
                # 更新访问次数和最后访问时间
                cursor.execute('''
                UPDATE stock_metadata 
                SET access_count = ?, last_accessed = datetime('now')
                WHERE id = ?
                ''', (access_count, stock_id))
                conn.commit()
                return stock_id

            # 创建新记录
            cursor.execute('''
            INSERT INTO stock_metadata 
            (user_input, mapped_code, stock_name, last_updated, access_count, last_accessed)
            VALUES (?, ?, ?, datetime('now'), 1, datetime('now'))
            ''', (user_input, mapped_code, stock_name))

            stock_id = cursor.lastrowid
            conn.commit()
            return stock_id

    def save_stock_data(self, stock_id, data):
        """保存K线数据"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()

            # 删除旧数据
            cursor.execute('DELETE FROM stock_data WHERE stock_id = ?', (stock_id,))

            # 插入新数据
            for index, row in data.iterrows():
                cursor.execute('''
                INSERT INTO stock_data 
                (stock_id, date, open, high, low, close, volume)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ''', (
                    stock_id,
                    index.strftime('%Y-%m-%d'),
                    row['Open'],
                    row['High'],
                    row['Low'],
                    row['Close'],
                    row['Volume'] if 'Volume' in row else 0
                ))

            # 更新最后更新时间
            cursor.execute('''
            UPDATE stock_metadata 
            SET last_updated = datetime('now') 
            WHERE id = ?
            ''', (stock_id,))

            conn.commit()

    def save_key_points(self, stock_id, key_points):
        """保存关键点数据"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()

            # 删除旧关键点
            cursor.execute('DELETE FROM key_points WHERE stock_id = ?', (stock_id,))

            # 插入新关键点
            for k in key_points:
                cursor.execute('''
                INSERT INTO key_points 
                (stock_id, date, point_type, price)
                VALUES (?, ?, ?, ?)
                ''', (
                    stock_id,
                    k['Date'].strftime('%Y-%m-%d'),
                    k['point_type'],
                    k['price']
                ))

            conn.commit()

    def load_stock_data(self, stock_id):
        """从缓存加载K线数据"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT date, open, high, low, close, volume 
            FROM stock_data 
            WHERE stock_id = ?
            ORDER BY date
            ''', (stock_id,))

            data = cursor.fetchall()
            if not data:
                return None

            # 转换为DataFrame
            df = pd.DataFrame(data, columns=['Date', 'Open', 'High', 'Low', 'Close', 'Volume'])
            df['Date'] = pd.to_datetime(df['Date'])
            df.set_index('Date', inplace=True)
            return df

    def load_key_points(self, stock_id):
        """从缓存加载关键点"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT date, point_type, price 
            FROM key_points 
            WHERE stock_id = ?
            ORDER BY date
            ''', (stock_id,))

            data = cursor.fetchall()
            if not data:
                return None

            key_points = []
            for row in data:
                key_points.append({
                    'Date': datetime.strptime(row[0], '%Y-%m-%d'),
                    'point_type': row[1],
                    'price': row[2]
                })

            return key_points

    def get_recent_stocks(self, limit=10):
        """获取最近访问的股票"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT id, user_input, mapped_code, stock_name 
            FROM stock_metadata 
            ORDER BY last_accessed DESC 
            LIMIT ?
            ''', (limit,))

            return cursor.fetchall()

    def get_most_popular_stocks(self, limit=10):
        """获取最常访问的股票"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT id, user_input, mapped_code, stock_name 
            FROM stock_metadata 
            ORDER BY access_count DESC 
            LIMIT ?
            ''', (limit,))

            return cursor.fetchall()

    def get_access_count(self, stock_id):
        """获取股票的访问次数"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT access_count FROM stock_metadata 
            WHERE id = ?
            ''', (stock_id,))

            result = cursor.fetchone()
            return result[0] if result else 0

    def clear_cache(self, older_than_days=30):
        """清除旧缓存"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()

            # 获取要删除的股票ID
            cursor.execute('''
            SELECT id FROM stock_metadata 
            WHERE date(last_accessed) < date('now', ?)
            ''', (f'-{older_than_days} days',))

            stock_ids = [row[0] for row in cursor.fetchall()]

            if not stock_ids:
                return 0

            # 删除相关数据
            cursor.execute(f'''
            DELETE FROM stock_data 
            WHERE stock_id IN ({','.join('?' * len(stock_ids))})
            ''', stock_ids)

            cursor.execute(f'''
            DELETE FROM key_points 
            WHERE stock_id IN ({','.join('?' * len(stock_ids))})
            ''', stock_ids)

            cursor.execute(f'''
            DELETE FROM stock_metadata 
            WHERE id IN ({','.join('?' * len(stock_ids))})
            ''', stock_ids)

            conn.commit()
            return len(stock_ids)

    def add_to_watchlist(self, stock_id):
        """添加股票到自选股"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            try:
                cursor.execute('''
                INSERT INTO watchlist (stock_id) 
                VALUES (?) 
                ON CONFLICT(stock_id) DO NOTHING
                ''', (stock_id,))
                conn.commit()
                return True
            except sqlite3.Error as e:
                logger.error(f"添加自选股失败: {str(e)}")
                return False

    def remove_from_watchlist(self, stock_id):
        """从自选股中移除股票"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            DELETE FROM watchlist 
            WHERE stock_id = ?
            ''', (stock_id,))
            conn.commit()
            return cursor.rowcount > 0

    def get_watchlist(self):
        """获取自选股列表"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT s.id, s.user_input, s.mapped_code, s.stock_name 
            FROM watchlist w
            JOIN stock_metadata s ON w.stock_id = s.id
            ORDER BY w.added_time DESC
            ''')
            return cursor.fetchall()

    def is_in_watchlist(self, stock_id):
        """检查股票是否在自选股中"""
        with self._db_lock:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute('''
            SELECT 1 FROM watchlist 
            WHERE stock_id = ?
            ''', (stock_id,))
            return cursor.fetchone() is not None

    def __del__(self):
        """关闭所有线程的连接"""
        try:
            # 关闭线程池执行器
            self.executor.shutdown(wait=False)

            # 关闭数据库连接
            if hasattr(self._thread_local, "conn") and self._thread_local.conn:
                self._thread_local.conn.close()
                self._thread_local.conn = None

            logger.info("缓存管理器资源已释放")
        except Exception as e:
            logger.error(f"释放缓存管理器资源时出错: {str(e)}")

    # 添加的异步方法
    def load_stock_data_async(self, stock_id, callback):
        """异步加载K线数据"""

        def load_task():
            try:
                data = self.load_stock_data(stock_id)
                callback(data)
            except Exception as e:
                logger.error(f"异步加载数据失败: {str(e)}")
                callback(None)

        self.executor.submit(load_task)

    def load_key_points_async(self, stock_id, callback):
        """异步加载关键点"""

        def load_task():
            try:
                key_points = self.load_key_points(stock_id)
                callback(key_points)
            except Exception as e:
                logger.error(f"异步加载关键点失败: {str(e)}")
                callback(None)

        self.executor.submit(load_task)

    def get_recent_stocks_async(self, limit, callback):
        """异步获取最近访问的股票"""

        def load_task():
            try:
                stocks = self.get_recent_stocks(limit)
                callback(stocks)
            except Exception as e:
                logger.error(f"异步获取最近股票失败: {str(e)}")
                callback([])

        self.executor.submit(load_task)

    def get_most_popular_stocks_async(self, limit, callback):
        """异步获取最常访问的股票"""

        def load_task():
            try:
                stocks = self.get_most_popular_stocks(limit)
                callback(stocks)
            except Exception as e:
                logger.error(f"异步获取热门股票失败: {str(e)}")
                callback([])

        self.executor.submit(load_task)

    def get_watchlist_async(self, callback):
        """异步获取自选股列表"""

        def load_task():
            try:
                watchlist = self.get_watchlist()
                callback(watchlist)
            except Exception as e:
                logger.error(f"异步获取自选股失败: {str(e)}")
                callback([])

        self.executor.submit(load_task)